package org.ns.learn.rpc.client;

import io.netty.channel.Channel;
import org.ns.learn.rpc.codec.RpcRequest;
import org.ns.learn.rpc.codec.RpcResponse;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicLong;

public class MessageManage {

    private Map<String, MessageCallback> messageCallbackMap = new ConcurrentHashMap<>(128);
    private AtomicLong id = new AtomicLong(0);
    public RpcClientResponse send(Channel channel, RpcRequest request) throws RpcRequestTimeouException{
        String requestId = id.incrementAndGet() + "";
        request.setRequestId(requestId);
//        request.s
        long requestTime = System.currentTimeMillis(); // 请求时间
        MessageCallback messageCallback = new MessageCallback(requestTime);
        messageCallbackMap.put(requestId, messageCallback);
        channel.writeAndFlush(request);
        RpcClientResponse response = messageCallback.getResponse(30);
        messageCallbackMap.remove(requestId);
        if (response == null && System.currentTimeMillis() - requestTime > 30000) {
            throw new RpcRequestTimeouException("请求超时");
        }
        return response;
    }

    public void setRpcResponse(RpcClientResponse response) {
        MessageCallback messageCallback = messageCallbackMap.remove(response.getRequestId());
        if (messageCallback != null) {
            messageCallback.setResponse(response);
        }
    }
}
